package com.netposa.datawarehouse.aggregator.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.netposa.appwebres.common.WebResInfo;
import com.netposa.appwebres.exception.WebResException;
import com.netposa.appwebres.extend.Page;
import com.netposa.datawarehouse.aggregator.client.fallback.BaseServiceClientFallback;
import com.netposa.datawarehouse.aggregator.client.search.VehicleResourceSearch;
import com.netposa.datawarehouse.aggregator.search.ResourceParam;
import com.netposa.datawarehouse.aggregator.search.ResourceResult;
import com.netposa.datawarehouse.aggregator.search.ResourceSearch;
import com.netposa.datawarehouse.aggregator.service.AggregatorService;
import com.netposa.datawarehouse.aggregator.service.helper.AggregatorServiceHelper;
import com.netposa.datawarehouse.aggregator.web.model.FtrCondition;
import com.netposa.datawarehouse.aggregator.web.model.ModuleTable;
import com.netposa.datawarehouse.ftr.DataSearchClient;
import com.netposa.datawarehouse.ftr.core.SearchDataEntry;
import com.netposa.datawarehouse.ftr.core.SearchDataRequest;
import com.netposa.datawarehouse.ftr.core.SearchDataResponse;
import com.netposa.datawarehouse.ftr.core.SearchParamEntry;
import com.netposa.datawarehouse.ftr.schema.IndexType;
import com.netposa.datawarehouse.ftr.schema.IndexTypes;
import com.netposa.library.api.IDictionaryCache;
import com.netposa.library.datawarehouse.aggregate.helper.ResourceDefined;
import com.netposa.library.datawarehouse.aggregate.helper.ResourceDefined.InterfaceType;
import com.netposa.library.datawarehouse.aggregate.helper.ResourceDefined.ResourceInnerFiled;
import com.netposa.library.datawarehouse.aggregate.helper.ResourceDefined.ResourceType;
import com.netposa.library.datawarehouse.aggregate.model.InterfaceInfo;
import com.netposa.library.datawarehouse.aggregate.utils.AggregatorUtils;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

@Service
public class AggregatorServiceImpl implements AggregatorService {
   private static final Logger log = LoggerFactory.getLogger(AggregatorServiceImpl.class);
   private final Logger logger = LoggerFactory.getLogger(AggregatorServiceImpl.class);
   @Autowired
   private VehicleResourceSearch vehicleResourceSearch;
   @Autowired
   private Map<String,ResourceSearch> resourceSearchMap;
   @Autowired
   private ModuleTable moduleTable;
   @Autowired
   private ApplicationContext applicationContext;
   @Autowired
   private BaseServiceClientFallback baseServiceClientFallback;
   @Autowired
   private IDictionaryCache dictionaryCache;

   public List fullTextRetrieval(String searchText, DataSearchClient searchClient) {
      SearchParamEntry entry = new SearchParamEntry();
      entry.setText(searchText);
      entry.setResultRange(new SearchParamEntry.ResultRange());
      SearchDataResponse response = this.fetchSearchData(entry, searchClient);
      return response.isSuccessful() ? response.getSources() : new ArrayList();
   }

   public List queryDataByType(ResourceDefined.ResourceType type, DataSearchClient searchClient) {
      IndexType indexTypes = null;
      switch (type) {
         case BODY:
            indexTypes = IndexTypes.BODY_INDEX_TYPE;
            break;
         case FACE:
            indexTypes = IndexTypes.FACE_INDEX_TYPE;
            break;
         case VEHICLE:
            indexTypes = IndexTypes.VEHICLE_INDEX_TYPE;
            break;
         case NONMOTOR:
            indexTypes = IndexTypes.NONMOTOR_INDEX_TYPE;
      }

      SearchParamEntry entry = new SearchParamEntry();
      entry.setResultRange(new SearchParamEntry.ResultRange(0, 100));
      SearchDataRequest request = new SearchDataRequest(indexTypes, entry);
      return searchClient.searchData(request).getSources();
   }

   public Page fullTextRetrievalWithAll(JSONObject jsonObject, DataSearchClient searchClient) {
      FtrCondition condition = JSON.parseObject(jsonObject.toJSONString(), FtrCondition.class);
      if (StringUtils.isNotBlank(condition.getPlateNumber())) {
         return this.getVehicleByPlate(condition);
      } else {
         List ftrResultDatas = new ArrayList();
         SearchParamEntry entry = new SearchParamEntry();
         entry.setText(condition.getSearchText() == null ? "" : condition.getSearchText());
         Integer from = (condition.getCurrentPage() - 1) * condition.getPageSize();
         Integer size = condition.getPageSize();
         entry.setResultRange(new SearchParamEntry.ResultRange(from, size));
         if (condition.getStartTime() != null && condition.getEndTime() != null) {
            entry.setTimeRange(new SearchParamEntry.TimeRange(condition.getStartTime(), condition.getEndTime()));
         }

         if (!StringUtils.isEmpty(condition.getSources()) && condition.getSources() != null) {
            entry.setSource(condition.getSources());
         }

         if (StringUtils.isNotEmpty(condition.getDeviceIds())) {
            SearchParamEntry.DeviceIdRange deviceIdRange = new SearchParamEntry.DeviceIdRange();
            deviceIdRange.addDeviceIds(condition.getDeviceIds());
            entry.setDeviceIdRange(deviceIdRange);
         }

         long startTimeEs = System.currentTimeMillis();
         log.warn("-------->开始从es检索");
         List entries = this.parallelFetchSearchData(entry, searchClient);
         log.warn("-------->结束从es检索，耗时：{}ms", System.currentTimeMillis() - startTimeEs);
         if (!entries.isEmpty()) {
            String plateSearch = null;
            if (StringUtils.isNotBlank(condition.getSearchText())) {
               plateSearch = AggregatorUtils.plateNumber(condition.getSearchText());
            }

            if (StringUtils.isNotEmpty(plateSearch)) {
               condition.setPlateNumber(plateSearch);
               ftrResultDatas = this.fetchSearchDataFromServers(entries, jsonObject, ResourceType.FACE, ResourceType.BODY, ResourceType.NONMOTOR);
               Pair vs = this.getVehicleByPlateX(condition);
               ftrResultDatas.addAll((Collection)vs.getRight());
            } else {
               long startTimeOpaq = System.currentTimeMillis();
               log.warn("-------->开始从opaq检索");
               ftrResultDatas = this.fetchSearchDataFromServers(entries, jsonObject);
               log.warn("-------->结束从opaq检索，耗时：{}ms", System.currentTimeMillis() - startTimeOpaq);
            }
         } else {
            this.logger.error("Fetch full text retrieval data failed condition {}. ", entry);
         }

         jsonObject.put("modules", this.moduleTable.getModuleTypes());
         Integer computeCount = this.fullTextCount(jsonObject);
         condition.updatePageInfo(ftrResultDatas, computeCount);
         return condition;
      }
   }

   private SearchDataResponse fetchSearchData(SearchParamEntry entry, DataSearchClient searchClient) {
      SearchDataRequest request = new SearchDataRequest(entry);
      return searchClient.searchData(request);
   }

   private List<SearchDataEntry> parallelFetchSearchData(SearchParamEntry entry, DataSearchClient searchClient) {
      List<SearchDataEntry> list = Lists.newCopyOnWriteArrayList();
      List<IndexType> indexTypes = Lists.newArrayList(IndexTypes.FACE_INDEX_TYPE, IndexTypes.BODY_INDEX_TYPE, IndexTypes.VEHICLE_INDEX_TYPE, IndexTypes.NONMOTOR_INDEX_TYPE);
      List<Callable<List<SearchDataEntry>>> tasks = Lists.newArrayList();
      indexTypes.forEach((indexType) -> {
         tasks.add(() -> {
            SearchDataRequest request = new SearchDataRequest(indexType, entry);
            SearchDataResponse response = searchClient.searchData(request);
            if (!response.isSuccessful()) {
               this.logger.error("Fetch full text retrieval data failed condition {} - {}. ", indexType, entry);
            }

            return response.getSources();
         });
      });
      ExecutorService service = null;

      try {
         service = Executors.newFixedThreadPool(tasks.size());
         service.invokeAll(tasks).forEach((f) -> {
            try {
               list.addAll(f.get());
            } catch (InterruptedException var3) {
               var3.printStackTrace();
            } catch (ExecutionException var4) {
               var4.printStackTrace();
            }

         });
      } catch (InterruptedException var11) {
         var11.printStackTrace();
      } finally {
         if (service != null) {
            service.shutdown();
         }

      }

      return list;
   }

   private FtrCondition getVehicleByPlate(FtrCondition condition) {
      List<ResourceResult> resourceResultList = this.vehicleResourceSearch.plateSearch(condition);
      Integer count = this.vehicleResourceSearch.count(condition);
      List resultList = Lists.transform(resourceResultList, (input) -> {
         return input.getData();
      });
      condition.updatePageInfo(resultList, count);
      return condition;
   }

   private Pair getVehicleByPlateX(FtrCondition condition) {
      List resourceResultList = this.vehicleResourceSearch.plateSearch(condition);
      Integer count = this.vehicleResourceSearch.count(condition);
      List resultList = Lists.transform(resourceResultList, ResourceResult::getData);
      return Pair.of(count, resultList);
   }

   public Integer fullTextCount(JSONObject condition) {
      Predicate<InterfaceInfo> isCountSelect = (i) -> {
         return i.getName().equals(InterfaceType.COUNT.getFiled());
      };
      Predicate<InterfaceInfo> isInType = (i) -> {
         return i.getType() != null && condition.getJSONArray("modules").contains(i.getType().getValue());
      };
      return this.moduleTable.getModules().parallelStream().mapToInt((moduleName) -> {
         AtomicInteger total = new AtomicInteger(0);
         this.moduleTable.getInterfaceInfo(moduleName).stream().filter(isInType).filter(isCountSelect).forEach((interfaceInfo) -> {
            try {
               WebResInfo webResInfo = this.baseServiceClientFallback.handler(interfaceInfo.getCurrentReqMethod(), moduleName, interfaceInfo.getCurrentPath(), JSON.toJSONString(condition));
               ResourceSearch resourceSearch = (ResourceSearch)this.applicationContext.getBean(interfaceInfo.getType().getValue());
               log.info("=======================>moduleName={}", moduleName);
               total.addAndGet(resourceSearch.count(webResInfo));
            } catch (Exception var7) {
               log.error("{}|{}|{}", "opaq统计模块记录总数异常", moduleName, var7);
            }

         });
         return total.intValue();
      }).reduce((left, right) -> {
         return left + right;
      }).orElse(0);
   }

   public Page<Map<String, Object>> selectAllInModules(JSONObject condition) {
      condition.put("deviceId", condition.get("deviceIds"));
      List<String> moduleTypeList = this.moduleTable.getModuleTypes();
      if (StringUtils.isNotBlank(condition.getString("searchText"))) {
         moduleTypeList.clear();
         moduleTypeList.addAll(AggregatorServiceHelper.perfectCondition(this.dictionaryCache, condition));
      }

      Map<String,Object> requestIdMap = new HashMap<>();
      String requestIds = condition.getString("requestId");
      if (StringUtils.isNotBlank(requestIds)) {
         requestIdMap = JSON.parseObject(requestIds, Map.class);
      }

      long startTimeOpaq = System.currentTimeMillis();
      log.warn("-------->开始从opaq检索");
      List<Callable<List<ResourceResult>>> tasks = Lists.newArrayList();
      Predicate<InterfaceInfo> isListSelect = (i) -> {
         return i.getName().equals(InterfaceType.LIST.getFiled());
      };
      Predicate<InterfaceInfo> isInType = (i) -> {
         return i.getType() != null && moduleTypeList.contains(i.getType().getValue());
      };

      Map<String, Object> finalRequestIdMap = requestIdMap;
      this.moduleTable.getModules().forEach((moduleName) -> {
         this.moduleTable.getInterfaceInfo(moduleName).stream().filter(isInType).filter(isListSelect).forEach((interfaceInfo) -> {
            tasks.add(() -> {
               String typeName = interfaceInfo.getType().getValue();
               ResourceSearch resourceSearch = (ResourceSearch)this.applicationContext.getBean(typeName);

               try {
                  if (MapUtils.isNotEmpty(finalRequestIdMap)) {
                     Object lastTime = finalRequestIdMap.get(typeName);
                     condition.put("lastAbsTime", lastTime);
                     if (typeName.equals(ResourceType.VEHICLE.getValue())) {
                        condition.put("lastPassTime", lastTime);
                     }
                  }

                  return  resourceSearch.selectInParams2(interfaceInfo.getCurrentReqMethod(), moduleName, typeName, interfaceInfo.getCurrentPath(), condition.toJSONString());
               } catch (Exception var8) {
                  log.error("{}|{}|{}", "opaq检索异常", moduleName, var8);
                  return Lists.newArrayList();
               }
            });
         });
      });
      long taskTime = System.currentTimeMillis();
      if (tasks.isEmpty()) {
         log.warn("-------->opaq检索任务为空，耗时：{}ms", taskTime - startTimeOpaq);
         return new Page<>();
      } else {
         List<ResourceResult> results = Lists.newArrayList();
         this.selectInPartition(results, tasks);
         long endTimeOpaq = System.currentTimeMillis();
         log.warn("-------->结束从opaq检索，耗时：{}ms", endTimeOpaq - taskTime);
         if (results.isEmpty()) {
            log.warn("-------->opaq检索结果为空，总耗时：{}ms", System.currentTimeMillis() - startTimeOpaq);
            return new Page<>();
         } else {
            condition.put("modules", moduleTypeList);
            Integer computeCount = this.fullTextCount(condition);
            log.warn("-------->opaq统计结果总数完成:{}，耗时：{}ms", computeCount, System.currentTimeMillis() - endTimeOpaq);
            List<Map<String, Object>> map = this.convertAndSortByTime(results, condition, requestIdMap);
            String conditionStr = condition.toJSONString();
            Page<Map<String, Object>> page = JSON.parseObject(conditionStr, Page.class);
            page.updatePageInfo(map, computeCount);
            return page;
         }
      }
   }

   private List fetchSearchDataFromServers(List entryList, JSONObject jsonObject) {
      return this.fetchSearchDataFromServers(entryList, jsonObject, ResourceType.values());
   }

   private List fetchSearchDataFromServers(List entryList, JSONObject jsonObject, ResourceDefined.ResourceType... resourceTypes) {
      Map<ResourceDefined.ResourceType,List<ResourceParam>> map = new HashMap(4);
      ResourceDefined.ResourceType[] var5 = resourceTypes;
      int var6 = resourceTypes.length;

      for(int var7 = 0; var7 < var6; ++var7) {
         ResourceDefined.ResourceType type = var5[var7];
         map.put(type, new ArrayList());
      }

      Iterator var10 = entryList.iterator();

      while(var10.hasNext()) {
         SearchDataEntry entry = (SearchDataEntry)var10.next();
         ResourceDefined.ResourceType resourceType = ResourceType.valueToResourceType(entry.getType());
         if (resourceType != null) {
            List params = map.get(resourceType);
            if (params != null && params.size() < 100) {
               ResourceParam resourceParam = new ResourceParam();
               resourceParam.setRecordId(entry.getRecordId());
               params.add(resourceParam);
            }
         }
      }

      List resourceResults = this.fetch(map);
      return this.convertAndSortByTime(resourceResults, jsonObject, null);
   }

   private List<Map<String, Object>> convertAndSortByTime(List<ResourceResult> results, JSONObject condition, Map<String, Object> requestIdMap) {
      Collections.sort(results, (o1, o2) -> {
         Object o1AbsTime = o1.getData().get(ResourceInnerFiled._ABS_TIME.getFiled());
         Object o2AbsTime = o2.getData().get(ResourceInnerFiled._ABS_TIME.getFiled());
         Long o1Long;
         if (o1AbsTime == null) {
            o1Long = 1L;
         } else {
            o1Long = this.getLongAbsTime(o1AbsTime);
         }

         Long o2Long;
         if (o2AbsTime == null) {
            o2Long = 1L;
         } else {
            o2Long = this.getLongAbsTime(o2AbsTime);
         }

         return Long.compare(o2Long, o1Long);
      });
      if (results.size() > (Integer)condition.get("pageSize")) {
         results = results.subList(0, (Integer)condition.get("pageSize"));
      }

      if (requestIdMap != null) {
         Iterator<ResourceResult> var4 = results.iterator();

         while(var4.hasNext()) {
            ResourceResult result = var4.next();
            String type = (String)result.getData().get(ResourceInnerFiled._TYPE.getFiled());
            Object absTime = result.getData().get(ResourceInnerFiled._ABS_TIME.getFiled());
            requestIdMap.put(type, absTime);
         }

         condition.put("requestId", JSON.toJSONString(requestIdMap));
      }

      return Lists.transform(results, ResourceResult::getData);
   }

   private Long getLongAbsTime(Object absTime) {
      if (absTime instanceof BigDecimal) {
         return ((BigDecimal)absTime).longValue();
      } else {
         return absTime instanceof Integer ? ((Integer)absTime).longValue() : (Long)absTime;
      }
   }

   public  List<ResourceResult> fetch(Map<ResourceDefined.ResourceType, List<ResourceParam>> map){
      List<Callable<List<ResourceResult>>> tasks = Lists.newArrayList();
      map.entrySet().forEach((entry) -> {
         tasks.add(() -> {
            ResourceDefined.ResourceType beanName = entry.getKey();
            List<ResourceParam> params = entry.getValue();
            if (params.isEmpty()) {
               return Lists.newArrayList();
            } else {
               ResourceSearch resourceSearch = this.resourceSearchMap.get(beanName.getValue());
               if (resourceSearch == null) {
                  this.logger.error("Not found bean {}.", beanName);
                  throw new WebResException("500");
               } else {
                  return resourceSearch.search(params);
               }
            }
         });
      });
      if (tasks.isEmpty()) {
         return Lists.newArrayList();
      } else {
         List<ResourceResult> results = Lists.newArrayList();
         this.selectInPartition(results, tasks);
         return results;
      }
   }

   private List<ResourceResult> selectInPartition(List<ResourceResult> results, List<Callable<List<ResourceResult>>> tasks) {
      ExecutorService executorService = null;

      try {
         executorService = Executors.newFixedThreadPool(tasks.size());
         List<Future<List<ResourceResult>>> futures = executorService.invokeAll(tasks);
         futures.stream().forEach((future) -> {
            try {
               results.addAll(future.get());
            } catch (InterruptedException e) {
               Thread.currentThread().interrupt();
               throw new RuntimeException(e);
            } catch (ExecutionException e) {
               Throwable cause = e.getCause() == null ? e : e.getCause();
               Throwables.propagateIfPossible(cause, RuntimeException.class);
               throw new RuntimeException(e);
            }
            /*try {
            } catch (Exception var3) {
               log.error("业务模块分块查询异常：" + var3);
               var3.printStackTrace();
            }*/

         });
      } catch (InterruptedException var8) {
         var8.printStackTrace();
      } finally {
         if (executorService != null) {
            executorService.shutdown();
         }

      }

      return results;
   }
}
